package com.ku.rocketmq.quickstart.filter;

import com.ku.rocketmq.constants.Const;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * @program: rocketmqDemo
 * @description: 主从模式 消息创建者
 * @author: JokerKu
 * @create: 2020-05-03 13:53
 **/
public class Provider {

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建默认的mq 并指定组名
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("test_quick_producer_name");
        //2.设置mq nameserver 地址
        defaultMQProducer.setNamesrvAddr(Const.NAMESERVER_ADDR);
        //3.启动生产者
        defaultMQProducer.start();

        //4.发送消息
        for (int i = 0; i < 5; i++) {
            //4.1创建消息
            Message message = new Message(
                    "test_quick_topic",//topic:主题
                    "TagA",//标签
                    "keyA" + i,//用户自定义的key 唯一标识
                    ("Hello RocketMQ" + i).getBytes()//要发送的消息内容实体(byte[])
            );
            message.putUserProperty("a",String.valueOf(i));

            //设置延时发送给consumer --> 延时消息
            //rocketMQ对于延时消息的设置
            //private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
            if (i == 1) {
                message.setDelayTimeLevel(2); //2对应5s
            }



            //4.2发送同步消息
            SendResult sendResult = defaultMQProducer.send(message);
            System.out.println("消息发送的结果:"+sendResult);

            //4.3发送异步消息
//            defaultMQProducer.send(message, new SendCallback() {
//                //消息发送成功
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    System.out.println("messageId:" + sendResult.getMsgId() + "--status:" + sendResult.getSendStatus());
//                }
//
//                //出现异常
//                @Override
//                public void onException(Throwable throwable) {
//                    System.out.println("发送失败");
//                }
//            });

            //4.4 指定topic中的队列发送消息
            SendResult send = defaultMQProducer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object arg) {
                    Integer queueNumber = (Integer) arg;
                    return list.get(queueNumber);

                }
            }, 2);//arg是队列编号 一个topic默认有4个队列 编号0-3
            System.out.println(send);
        }

        //5.关闭生产者
        //defaultMQProducer.shutdown();

    }
}
